/*
 * Copyright (c) 2006-2021, RT-Thread Development Team
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * Change Logs:
 * Date           Author       Notes
 * 2015-07-29     Arda.Fu      first implementation
 */
#include <stdint.h>
#include <stdlib.h>
#include <string.h>

#include <rtthread.h>
#include <rtdevice.h>
#include "gps_rmc.h"
#include "paho_mqtt.h"
#ifndef GPS_RMC_SAMPLE_UART_NAME
#define GPS_RMC_SAMPLE_UART_NAME "uart2"
#endif // !GPS_RMC_SAMPLE_UART_NAME

/* 定义邮箱控制块 */
static rt_mailbox_t msg_mail = RT_NULL;

#define MQTT_TEST_SERVER_URI    "tcp://xxx.xxx.xxx.xxx:1883"//修改为自己的IP地址
#define MQTT_CLIENTID           "rtthread-mqtt"
#define MQTT_USERNAME           "admin"
#define MQTT_PASSWORD           "admin"
#define MQTT_SUBTOPIC           "/mqtt/warning"
#define MQTT_PUBTOPIC           "/mqtt/location"
#define MQTT_WILLMSG            "Goodbye!"
#define MQTT_TEST_QOS           1
#define MQTT_PUB_SUB_BUF_SIZE   1024


#define TEST_DATA_SIZE          256
#define PUB_CYCLE_TM            1000

static rt_thread_t pub_thread_tid = RT_NULL;

static char *pub_data = RT_NULL;

/* define MQTT client context */
static MQTTClient client;

static rt_uint32_t pub_count = 0, sub_count = 0;
static int recon_count = -1;
static int test_start_tm = 0;
static int test_is_started = 0;

/* 鐢ㄤ簬鎺ユ敹娑堟伅鐨勪俊鍙烽噺 */
static struct rt_semaphore rx_sem;
static rt_device_t uart;

/* 鎺ユ敹鏁版嵁鍥炶皟鍑芥暟 */
static rt_err_t uart_input(rt_device_t dev, rt_size_t size)
{
    /* 涓插彛鎺ユ敹鍒版暟鎹悗浜х敓涓柇锛岃皟鐢ㄦ鍥炶皟鍑芥暟锛岀劧鍚庡彂閫佹帴鏀朵俊鍙烽噺 */
    rt_sem_release(&rx_sem);

    return RT_EOK;
}

void gps_rmc_sample_entry(void *p)
{
    char buff[128] = {0}, *buff_p = buff, ch = 0;
    struct gps_info info_data = {0};
    gps_info_t info = &info_data;
    rt_err_t uwRet = RT_EOK;
    while (1)
    {
        /* 浠庝覆鍙ｈ鍙栦竴涓瓧鑺傜殑鏁版嵁锛屾病鏈夎鍙栧埌鍒欑瓑寰呮帴鏀朵俊鍙烽噺 */
        while (rt_device_read(uart, -1, &ch, 1) != 1)
        {
            /* 闃诲绛夊緟鎺ユ敹淇″彿閲忥紝绛夊埌淇″彿閲忓悗鍐嶆璇诲彇鏁版嵁 */
            rt_sem_take(&rx_sem, RT_WAITING_FOREVER);
        }
        if (ch == '\n')
        {
            if (rt_strstr((const char *)buff, "RMC"))
            {
                if (gps_rmc_parse(info, buff))
                {
                    char temp[64];
                    gps_print_info(info);
                    sprintf(temp,"{\"lon\":%0.6f,\"lat\":%06f}",info->coord.location.longitude.value,info->coord.location.latitude.value);
                    uwRet = rt_mb_send(msg_mail,(rt_uint32_t)&temp);
                                if (RT_EOK == uwRet)
                                    rt_kprintf ( "邮箱消息发送成功\n" );
                                else
                                    rt_kprintf ( "邮箱消息发送失败\n" );
                }
            }
            rt_memset(buff, 0, 128);
            rt_memset(info, 0, sizeof(struct gps_info));
            buff_p = buff;
            continue;
        }
        *buff_p++ = ch;
    }
}

int gps_rmc_sample_entry_init(void)
{

    uart = rt_device_find(GPS_RMC_SAMPLE_UART_NAME);
    if (uart == RT_NULL)
    {
        rt_kprintf("Not find %s device.\r\n", GPS_RMC_SAMPLE_UART_NAME);
        return RT_ERROR;
    }

    /* 鍒濆鍖栦俊鍙烽噺 */
    rt_sem_init(&rx_sem, GPS_RMC_SAMPLE_UART_NAME"_rx", 0, RT_IPC_FLAG_FIFO);

    struct serial_configure config = RT_SERIAL_CONFIG_DEFAULT;
    config.baud_rate = BAUD_RATE_9600;
    rt_device_control(uart, RT_DEVICE_CTRL_CONFIG, &config);
    rt_device_open(uart, RT_DEVICE_FLAG_INT_RX);
    rt_device_set_rx_indicate(uart, uart_input);

    rt_thread_t t = rt_thread_create(
        "gps_rmc_p", gps_rmc_sample_entry, RT_NULL,
        2048, 20, 10
    );
    if (t == RT_NULL)
    {
        rt_kprintf("Failde to create gps rmc info procees thread.\r\n");
        return RT_ERROR;
    }
    if (rt_thread_startup(t) != RT_EOK)
    {
        rt_kprintf("Failde to startup gps rmc info procees thread.\r\n");
        rt_thread_delete(t);
        return RT_ERROR;
    }
    return RT_EOK;
}

static void mqtt_sub_callback(MQTTClient *c, MessageData *msg_data)
{
    sub_count ++;
    rt_kprintf("payload   : %s\n", msg_data->message->payload);
    return;
}

static void mqtt_connect_callback(MQTTClient *c)
{
    return;
}

static void mqtt_online_callback(MQTTClient *c)
{
    recon_count ++;
    return;
}

static void mqtt_offline_callback(MQTTClient *c)
{
    return;
}

/**
 * This function create and config a mqtt client.
 *
 * @param void
 *
 * @return none
 */
static void mq_start(void)
{
    /* init condata param by using MQTTPacket_connectData_initializer */
    MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;

    rt_memset(&client, 0, sizeof(MQTTClient));

    /* config MQTT context param */
    {
        client.uri = MQTT_TEST_SERVER_URI;

        /* config connect param */
        memcpy(&client.condata, &condata, sizeof(condata));
        client.condata.clientID.cstring = MQTT_CLIENTID;
        client.condata.keepAliveInterval = 60;
        client.condata.cleansession = 1;
        client.condata.username.cstring = MQTT_USERNAME;
        client.condata.password.cstring = MQTT_PASSWORD;

        /* config MQTT will param. */
        client.condata.willFlag = 1;
        client.condata.will.qos = MQTT_TEST_QOS;
        client.condata.will.retained = 0;
        client.condata.will.topicName.cstring = MQTT_PUBTOPIC;
        client.condata.will.message.cstring = MQTT_WILLMSG;

        /* malloc buffer. */
        client.buf_size = client.readbuf_size = MQTT_PUB_SUB_BUF_SIZE;
        client.buf = rt_malloc(client.buf_size);
        client.readbuf = rt_malloc(client.readbuf_size);
        if (!(client.buf && client.readbuf))
        {
            rt_kprintf("no memory for MQTT client buffer!\n");
            goto _exit;
        }

        /* set event callback function */
        client.connect_callback = mqtt_connect_callback;
        client.online_callback = mqtt_online_callback;
        client.offline_callback = mqtt_offline_callback;

        /* set subscribe table and event callback */
        client.messageHandlers[0].topicFilter = rt_strdup(MQTT_SUBTOPIC);
        client.messageHandlers[0].callback = mqtt_sub_callback;
        client.messageHandlers[0].qos = MQTT_TEST_QOS;

        /* set default subscribe event callback */
        client.defaultMessageHandler = mqtt_sub_callback;
    }

    /* run mqtt client */
    paho_mqtt_start(&client);

    return;

_exit:
    if (client.buf)
    {
        rt_free(client.buf);
        client.buf = RT_NULL;
    }
    if (client.readbuf)
    {
        rt_free(client.readbuf);
        client.readbuf = RT_NULL;
    }
    return;
}

static void show_test_info(void)
{
    char temp[50] = {0};
    rt_kprintf("\r==== MQTT Stability test ====\n");
    rt_kprintf("Server: "MQTT_TEST_SERVER_URI"\n");
    rt_kprintf("QoS   : %d\n", MQTT_TEST_QOS);

    //rt_kprintf("Test duration(sec)            : %d\n", time((time_t *)RT_NULL) - test_start_tm);
    rt_kprintf("Number of published  packages : %d\n", pub_count);
    rt_kprintf("Number of subscribed packages : %d\n", sub_count);
    snprintf(temp, sizeof(temp), "Packet loss rate              : %.2f%%   \n", (float)((float)(pub_count - sub_count) * 100.0f / pub_count));
    rt_kprintf(temp);
    rt_kprintf("Number of reconnections       : %d\n", recon_count);
    /* up the cursor 8 line */
    rt_kprintf("\033[8A");
}

static void thread_pub(void *parameter)
{
    pub_data = rt_malloc(TEST_DATA_SIZE * sizeof(char));
    if (!pub_data)
    {
        rt_kprintf("no memory for pub_data\n");
        return;
    }
    rt_memset(pub_data, '*', TEST_DATA_SIZE * sizeof(char));

    test_start_tm = time((time_t *)RT_NULL);
    rt_kprintf("test start at '%d'\r\n", test_start_tm);
    rt_err_t uwRet = RT_EOK;
       char *r_str;
    while (1)
    {
        char temp [64];


        //rt_thread_delay(PUB_CYCLE_TM);

        //show_test_info();
        /* 等待接邮箱消息 */
               uwRet = rt_mb_recv(msg_mail, /* 邮箱对象句柄 */
                               (rt_uint32_t *)&r_str, /* 接收邮箱消息 */
                               RT_WAITING_FOREVER);/* 指定超时事件,一直等 */

               if (RT_EOK == uwRet) { /* 如果接收完成并且正确 */
                   rt_kprintf ( "邮箱的内容是:%s\n\n",r_str);
                   if (!paho_mqtt_publish(&client, QOS1, MQTT_PUBTOPIC, r_str))
                   {
                       ++ pub_count;
                   }
               } else
                   rt_kprintf ( "邮箱接收错误！错误码是0x%x\n",uwRet);

    }
}

static void mqtt_test_start(void)
{
    if (test_is_started)
    {
        return;
    }

    mq_start();

    while (!client.isconnected)
    {
        rt_kprintf("Waiting for mqtt connection...\n");
        rt_thread_delay(1000);
    }

    pub_thread_tid = rt_thread_create("pub_thread", thread_pub, RT_NULL, 1024, 8, 100);
    if (pub_thread_tid != RT_NULL)
    {
        rt_thread_startup(pub_thread_tid);
    }

    test_is_started = 1;

    return;
}


/* defined the LED1 pin: PB5 */
#define LED1_PIN    57

int main(void)
{
    uint32_t Speed = 200;
    /* 创建一个邮箱 */
        msg_mail = rt_mb_create("msg_mail", /* 邮箱名字 */
                                10,         /* 邮箱大小 */
                                RT_IPC_FLAG_FIFO);/* 信号量模式 FIFO(0x00)*/
        if (msg_mail != RT_NULL)
                rt_kprintf("邮箱创建成功！\n\n");
    /* set LED1 pin mode to output */
    rt_pin_mode(LED1_PIN, PIN_MODE_OUTPUT);
    gps_rmc_sample_entry_init();
    mqtt_test_start();
    while (1)
    {
        rt_pin_write(LED1_PIN, PIN_LOW);
        rt_thread_mdelay(Speed);
        rt_pin_write(LED1_PIN, PIN_HIGH);
        rt_thread_mdelay(Speed);
    }
}

